package com.dhl.rabbitmq.four;

import com.dhl.rabbitmq.utils.RabbitUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布消息确认机制
 * @title: Task2
 * @Author danghailiang
 * @Date: 2022/12/7 10:34
 * @Version 1.0
 */
public class ConfirmMessageTest {


    public static void main(String[] args) throws Exception {
        // 当前单个应答消息发送消息:1000个，消耗的时间为：1302ms
        onePublishMessage();
        // 当前多个应答消息发送:1000个，消耗的时间为：231ms
        multiPublishMessage();
        // 当前多个应答异步消息发送:1000个，消耗的时间为：47ms
        asyncPublishMessage();
    }

    /**
     * 每次单个消息发布确认
     * @throws Exception
     */
    public static void onePublishMessage()   {
        int messageCount = 1000;
        // 获取信道
        try ( Channel channel = RabbitUtils.getChannel()) {
            // 设置信道名称
            String queueName = UUID.randomUUID().toString();
            /**
             * 队列名称
             * 是否持久化
             * 是否只提供一个消费者消费，既：是否消息共享 （true）不进行共享 false消息共享
             * 是否自动删除
             */
            channel.queueDeclare(queueName,true,false,false,null);
            // 开启消息确认
            channel.confirmSelect();
            // 开始时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i< messageCount; i++){
                String message = "消息："+i;
                channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));
                // 消息确认
                boolean confirmsFlag = channel.waitForConfirms();
                if(confirmsFlag){
                    System.out.println("消息确认成功：" + i );
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("当前单个应答消息发送消息:"+ messageCount +"个，消耗的时间为："+( end - begin ) + "ms");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 多个消息批量确认
     * @throws Exception
     */
    public static void multiPublishMessage()   {
        int messageCount = 1000;
        // 获取信道
        try ( Channel channel = RabbitUtils.getChannel()) {
            // 设置信道名称
            String queueName = UUID.randomUUID().toString();
            /**
             * 队列名称
             * 是否持久化
             * 是否只提供一个消费者消费，既：是否消息共享 （true）不进行共享 false消息共享
             * 是否自动删除
             */
            channel.queueDeclare(queueName,true,false,false,null);
            // 开启消息确认
            channel.confirmSelect();
            // 开始时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i< messageCount; i++){
                String message = "消息："+i;
                channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));
                if(i%50==0){
                    // 消息确认
                    boolean confirmsFlag = channel.waitForConfirms();
                    if(confirmsFlag){
                        System.out.println("消息确认成功:"+ i);
                    }
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("当前单个应答消息发送消息:"+ messageCount +"个，消耗的时间为："+( end - begin ) + "ms");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 异步确认
     * @throws Exception
     */
    public static void asyncPublishMessage()   {
        int messageCount = 1000;
        // 获取信道
        try ( Channel channel = RabbitUtils.getChannel() ) {
            // 开启消息确认
            channel.confirmSelect();
            // 设置信道名称
            String queueName = UUID.randomUUID().toString();
            /**
             * 队列名称
             * 是否持久化
             * 是否只提供一个消费者消费，既：是否消息共享 （true）不进行共享 false消息共享
             * 是否自动删除
             */
            channel.queueDeclare(queueName,true,false,false,null);
            // 消息确认集合，标记消息使用
            ConcurrentSkipListMap<Long, String> concurrentSkipListMap = new ConcurrentSkipListMap();
            // 消息确认成功回调
            ConfirmCallback ackCallback = (deliveryTag, multiple) ->{
                // 是否开启了批量处理
                if(multiple){
                    // 返回的是小于等于当前序列号的未确认消息 是一个map
                    ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap = concurrentSkipListMap.headMap(deliveryTag, true);
                    //清除该部分未确认消息
                    longStringConcurrentNavigableMap.clear();
                }else{
                    // 删除发送成功的消息
                    concurrentSkipListMap.remove(deliveryTag);
                }
                System.out.println("异步确认消息：" + deliveryTag);
            };
            // 消息确认失败回调
            ConfirmCallback nackCallback = (deliveryTag, multiple) ->{
                // 获取未成功发送成功的消息
                String message = concurrentSkipListMap.get(deliveryTag);
                System.out.println("异步未确认消息编号：" + deliveryTag +"，消息为：" + message);
            };
            // 添加异步消息监听
            channel.addConfirmListener(ackCallback,nackCallback);
            // 开始时间
            long begin = System.currentTimeMillis();
            for (int i = 0; i< messageCount; i++){
                String message = "消息："+i;
                // 获取下一条消息序号和消息，进行保存
                concurrentSkipListMap.put(channel.getNextPublishSeqNo(),message);
                // 发送消息
                channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));

            }
            long end = System.currentTimeMillis();
            System.out.println("当前应答消息异步确认:"+ messageCount +"个，消耗的时间为："+( end - begin ) + "ms");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
